package com.atguigu.gmall.realtime.app.dwm;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.text.SimpleDateFormat;

/**
 * @Description: dwm层 访问uv计算 (独立访客计算),对于实时统计,也可以理解为 DAU,也就是日活.
 * @Author: tiancy
 * @Create: 2021/11/29
 */

/*
    做实时的情况下,我们从中间分出来一层 dwm层,这里可能存在争议,因为中间加入一层 `会对我们的时效性存在影响`.
    这里我们加入一层DWM,主要服务dws层. 主要对实时数据做一个轻量聚合. 也考虑到复用性(计算结果复用) ==> 放到olap中,如果开始统计为1000条,经过轻量聚合,可能只剩下1条.
    并且,我们的维度数据会放到HBase中,有些计算的数据可以直接在ods层直接拿到,有些却需要我们组合.
    比如 下单主题 需要 订单表 + 订单明细表 + 维度(用户 + 商品 + 地区 + 品牌 ....) 组合的宽表,再计算分析 .

 */

/*
    日活需求分许
    数据来源 : dwd_page_log(Kafka主题)
 */
public class UniqueVisitorApp {
    public static void main(String[] args) throws Exception {
        /*
            TODO 处理步骤
                1、基本环境准备 [设置流环境、设置并行度]
                2、检查点的设置
                3、从Kafka主题中读取数据 . 声明消费主题以及消费者组 . 获取一个消费者对象构建的流.
                4、对流中的数据进行格式转换 json字符串转化为json对象.
                5、按照用户标识分组
                6、去重
                7、将去重之后的独立访客,写法哦Kafka的dwm主题中.
                需要起的服务 : HDFS、zk、Kafka、数据来源部分==>用户行为日志部分 `rt_logger.sh start (日志采集服务 + Nginx)`、`/opt/module/applog_realtime/gmall2020-mock-log-2020-12-18.jar)`
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2.检查点相关设置
        //2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置取消job之后是否保留检查点
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 3000L));
        //2.5 设置状态后端
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/gmall/dbCK"));
        //2.6 设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME", "atguigu");

        String topicName = "dwd_page_log";
        String groupId = "unique_visitor_app_group";

        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topicName, groupId);
        DataStreamSource<String> KafkaDS = env.addSource(kafkaSource);

        SingleOutputStreamOperator<JSONObject> jsonObjDS = KafkaDS.map(JSON::parseObject);

       /* jsonObjDS.keyBy(new KeySelector<JSONObject, Object>() {
            @Override
            public Object getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getJSONObject("common").getString("mid");
            }
        });*/
        KeyedStream<JSONObject, String> keyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));
        /** 使用状态进行去重*/
        SingleOutputStreamOperator<JSONObject> filterDS = keyedDS.filter(new RichFilterFunction<JSONObject>() {

            private ValueState<String> midState;
            private SimpleDateFormat sdf;

            @Override
            public void open(Configuration parameters) throws Exception {
                sdf = new SimpleDateFormat("yyyyMMdd");
                ValueStateDescriptor<String> midValueState = new ValueStateDescriptor<>("midValueState", Types.STRING);
                // 设置状态的失效时间.
                StateTtlConfig ttlConfig = StateTtlConfig
                        .newBuilder(Time.days(1))  // 声明状态从声明开始,过期时间为1天
                        /*
                            setUpdateType : 此选项值配置 何时更新延长状态TTL的上次访问时间戳.
                                - Disabled : 状态过期禁用,就是永不过期.
                                - OnCreateAndWrite : 上次访问时间戳在每次写入操作时创建和更新状态时初始化。
                                - OnReadAndWrite : 与 OnCreateAndWrite相同，但也在读取时更新。
                         */
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        /*
                            此选项配置是否可以返回过期的用户值
                                ReturnExpiredIfNotCleanedUp : 如果尚未清除，则返回过期的用户值
                                NeverReturnExpired : 永不返回过期的用户值
                         */
                        .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                        .build();
                midValueState.enableTimeToLive(ttlConfig);
                midState = getRuntimeContext().getState(midValueState);
            }

            // 主要用来判断什么数据保留,什么数据直接舍弃.
            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                // 获取当前页面的上一页id,如果上一页的id为 null,则为首页.如果存在上一页,则直接过滤掉.
                String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                /* if (lastPageId == null) return false;*/
                if (StringUtils.isNotEmpty(lastPageId)) return false;

                /** 利用状态保存历史访问的日期，判断是否曾经访问过 */
                //从状态中获取上次访问日期
                String lastVisitDate = midState.value();
                //获取当前日志访问日期
                String curDate = sdf.format(jsonObject.getLong("ts"));

                //判断今天是否访问过,也就是一天访问多次. 存在状态变量 并且 当前状态中的日期 == 当前行数据中的ts
                if (lastVisitDate != null && lastVisitDate.length() > 0 && curDate.equals(lastVisitDate)) {
                    //已经访问过  直接将再次访问的数据给过滤掉
                    return false;
                } else {
                    //没有访问过 ，将当前访问的日期放到状态中保存起来
                    midState.update(curDate);
                    return true;
                }
            }
        });
        filterDS.print("dwm_unique_visitor >>>");
        filterDS
                .map(JSON::toString)
                .addSink(MyKafkaUtil.getKafkaSink("dwm_unique_visitor"));
        env.execute();
    }
}
